package onion.mqtt.server.processor;

import io.netty.channel.Channel;
import io.netty.channel.ChannelFutureListener;
import io.netty.handler.codec.mqtt.MqttMessage;
import io.netty.util.AttributeKey;
import onion.mqtt.server.MqttServerConst;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * @author Mr, Lu
 * @developmentTeam 浙江允泽信息科技有限公司
 * @createTime 2023/12/13
 */
public abstract class AbstractMqttServerProcessor<T extends MqttMessage> {
    static final Logger log = LoggerFactory.getLogger(AbstractMqttServerProcessor.class);
    /**
     * 处理消息
     *
     * @param channel
     * @param message
     */
    public abstract void process(Channel channel, T message);

    /**
     * 获取ClientId
     *
     * @param channel
     * @return
     */
    public String getClientId(Channel channel) {
        return (String) channel.attr(AttributeKey.valueOf(MqttServerConst.CLIENT_ID)).get();
    }

    /**
     * 异步写消息
     *
     * @param channel
     * @param message
     */
    public void writeAndFlush(Channel channel, MqttMessage message) {
        channel.eventLoop().execute(() -> channel.writeAndFlush(message));
    }

    /**
     * 关闭channel
     *
     * @param channel
     */
    public synchronized void close(Channel channel) {
        channel.close().addListener((ChannelFutureListener) future -> {
            log.debug("channel close clientId: {}", channel.attr(AttributeKey.valueOf(MqttServerConst.CLIENT_ID)).get());
            if (!future.isSuccess()) {
                log.error("channel close error ");
            }
        });
    }
}
